package chapter5.eg7;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * @author zhangxufeng@meitunmama.com 2019-09-02 21:33:13
 */
public class Producer {

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.ACKS_CONFIG, "all");
    properties.put(ProducerConfig.RETRIES_CONFIG, 3);
    properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840);
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);
    properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
    properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);

    KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
    Future<RecordMetadata> future = producer
        .send(new ProducerRecord<>("test-topic", "testKey", "testValue"));
    RecordMetadata metadata = future.get();
    System.out.println(
        "topic: " + metadata.topic() + ", partition: " + metadata.partition() + ", offset: "
            + metadata.offset());
  }

}
